package com.tunnel.common.tunnel;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.tunnel.common.util.LogUtil;
import com.tunnel.common.util.StringUtil;

import io.netty.channel.ChannelHandlerContext;

/**
 * tunnel 数据管道
 * 阻塞队列，消费者模式
 */
public class TunnelDataQueue {

	//tunnel通道的数据包大小设定
	private volatile static  int ITEM_SIZE = 100;//默认100b，初始化配置时候，如果低于100无效
	//单位换算，key是单位,value是byte的倍数
	private static final Map<String,Integer> UNIT_MAP = new HashMap<>();
	static{
		UNIT_MAP.put("B", 1);
		UNIT_MAP.put("K", 1000);
		UNIT_MAP.put("M", 1000*1000);
	}
	//窗口期的时间长度
	public static final long ITEM_SEND_DELAY_TIME = 10l;//单位毫秒，默认每10毫秒发送不超过ITEM_SIZE大小的数据包，次数可能不只1次
	
	//*****************双队列，读写分离*******************//
	//使用LinkedQuee，因为考虑吞吐速度可以慢一点，但是对空间利用率希望好一些
	private List<TunnelData> READ_QUEUE = new LinkedList<>();
	private List<TunnelData> WRITE_QUEUE = new LinkedList<>();
	private volatile static int QUEUE_SIZE = 500;//队列长度500，这样可以减少最长的等待延迟
	//*****************
	//这个可以再Channel初始化后，连接成功后就添加到这里来，以便心跳包的发送
	private String name;
	private ChannelHandlerContext ctx;
	private boolean needBeatHeart = false;//是否需要心跳包维持
	
	public ChannelHandlerContext getCtx() {
		return ctx;
	}

	private volatile boolean stop = false;
	
	public TunnelDataQueue(String name,ChannelHandlerContext ctx,boolean needBeatHeart) {
		this.name = name;
		this.ctx = ctx;
		this.needBeatHeart = needBeatHeart;
	}
	
	public void stop(){
		synchronized (READ_QUEUE) {
			synchronized (WRITE_QUEUE) {
				READ_QUEUE.clear();
				WRITE_QUEUE.clear();
			}
		}
		stop = true;
	}

	public static void setITEM_SIZE(String speed) throws Exception {
		Set<String> unitSet = UNIT_MAP.keySet();
		String unitStr = "";
		for(String unit:unitSet){
			unitStr = unitStr+unit;
		}
		
		if(!StringUtil.checkStr(speed.toUpperCase(), "^([1-9]+[0-9]*)+(\\.[0-9]*[1-9])?["+unitStr+"]{1}$")){
			throw new Exception("无效的设置："+speed);
		}
		BigDecimal itemSize_bd = new BigDecimal(speed.substring(0, speed.length()-1));
		BigDecimal ratio_bd = new BigDecimal(UNIT_MAP.get(speed.substring(speed.length()-1).toUpperCase()));
		//因为ITEM_SEND_DELAY_TIME的单位是毫秒，但是配置文件是单位秒，所以这里要*1000作为除数，来计算每毫秒上传的每块数据包大小上限
		ITEM_SIZE = itemSize_bd.multiply(ratio_bd).multiply(new BigDecimal(ITEM_SEND_DELAY_TIME)).divide(new BigDecimal(1000),0,BigDecimal.ROUND_DOWN).intValue();
		
		
		LogUtil.d(">>>>>> TunnelDataQueue speed: "+ITEM_SIZE+"b/"+TunnelDataQueue.ITEM_SEND_DELAY_TIME+"ms <<<<<<<");
	}
	
	public static int ITEM_SIZE() {
		return ITEM_SIZE;
	}
	
	public void startQueue(){
		new Thread("TunnelDataQueue-"+name){
			public void run() {
				//数据发送的窗口时间
				long sendDataWindowTime = System.nanoTime();
				//数据发送的窗口大小，这是个滑块变量，会动态调整
				int sendDataWindowSize = ITEM_SIZE;
				//心跳包的发送时间周期1秒
				long beatHeartDelayTime = 1000*1000000;
				//心跳包的发送时间
				long sendBeatHeartTime = sendDataWindowTime;
				while(!stop){
					try {
						TunnelData tunnelData = null;
						synchronized (READ_QUEUE) {
							long now = System.nanoTime();
							if(now-sendDataWindowTime<ITEM_SEND_DELAY_TIME*1000*1000){//因为nanoTime单位是纳秒，是1000000毫秒
								//窗口期还没到，必须检查窗口大小
								if(sendDataWindowSize<=0){
									//如果窗口大小已经没有剩余，窗口已关闭
//									LogUtil.d("waite:"+sendWindowTime+"/"+sendWindowSize);
									Thread.sleep(1);//休息1毫秒，防止cpu空转
									continue;
								}
								/*if(sendWindowSize != ITEM_SIZE){
									LogUtil.d("tunnel sendWindowSize:"+sendWindowSize);
								}*/
							}else{
								//时间到了，更新窗口参数
								sendDataWindowTime = now;
								sendDataWindowSize = ITEM_SIZE;
							}
//							LogUtil.d("send:"+sendWindowTime+"/"+sendWindowSize);
							
							if(READ_QUEUE.size() > 0){
								tunnelData = READ_QUEUE.remove(0);
							}else{
								//如果R队列空了，则发一个心跳包，并且交换RW队列
								if(ctx != null && needBeatHeart && now-sendBeatHeartTime>beatHeartDelayTime){
									tunnelData = new TunnelData(TunnelDataSenderManager.PING_MSG);
									//更新心跳包发送时间
									sendBeatHeartTime = now;
									//如果是发送心跳包，那么本次窗口期，必须占满，以防止cpu占用过高
									sendDataWindowSize = 0;
								}
								synchronized (WRITE_QUEUE) {
									if(READ_QUEUE.size() == 0 && WRITE_QUEUE.size() > 0){
//											LogUtil.d("交换RW队列0->"+WRITE_QUEUE.size());
										List<TunnelData> tmpQueue = WRITE_QUEUE;
										WRITE_QUEUE = READ_QUEUE;
										READ_QUEUE = tmpQueue;
									}
								}
							}
						}
						if(tunnelData != null){
							/*if(tunnelData.getDatas().length > 0){
								LogUtil.d("tunnel sendWindowTime:"+sendWindowTime/1000/1000+"ms");
							}*/
							
							TunnelDataSenderManager.doSendData(ctx,tunnelData);
							//只要有数据发送，就需要跟新窗口剩余大小int dataLen = 0;
			            	for(byte[] dataOne:tunnelData.getDatas()){
			            		if(dataOne != null){
									sendDataWindowSize = sendDataWindowSize-dataOne.length;
			            		}
			            	}
						}else{
							Thread.sleep(1);//休息1毫秒，防止cpu空转
						}
					} catch (Exception e) {
						LogUtil.e(e);
					}
				}
//				LogUtil.d("队列结束");
			};
		}.start();
	}
	
	public boolean addDataToQueue(TunnelData tunnelData){
		synchronized (WRITE_QUEUE) {
			if(WRITE_QUEUE.size() >= QUEUE_SIZE){
				return false;
			}
			WRITE_QUEUE.add(tunnelData);
			return true;
		}
	}
	
	public static void main(String[] args) {
		try {
			setITEM_SIZE("2M");
		} catch (Exception e) {
			// TODO: handle exception
		}
	}
}
